package main

import (
	"context"
	"fmt"
	"grpcPrimary/pb"
	"io"
	"log"
	"time"

	epb "google.golang.org/genproto/googleapis/rpc/errdetails"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/metadata"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/types/known/wrapperspb"
)

const (
	address = "localhost:8080"
)

func orderUnaryClientInterceptor(
	ctx context.Context,
	method string,
	req, reply interface{},
	cc *grpc.ClientConn,
	invoker grpc.UnaryInvoker,
	opts ...grpc.CallOption) error {
	// 前置处理阶段
	log.Println("Method : " + method)

	// 调用远程方法
	err := invoker(ctx, method, req, reply, cc, opts...)

	// 后置处理阶段
	log.Println(reply)

	return err
}

type wrappedSteam struct {
	grpc.ClientStream
}

func (w *wrappedSteam) RecvMsg(m interface{}) error {
	log.Printf("====== [Client Stream Interceptor] "+
		"Reveive a message (Type: %T) at %v", m, time.Now().Format(time.RFC3339))
	return w.ClientStream.RecvMsg(m)
}

func (w *wrappedSteam) SendMsg(m interface{}) error {
	log.Printf("====== [Client Stream Interceptor] "+
		"Send a message (Type: %T) at %v", m, time.Now().Format(time.RFC3339))
	return w.ClientStream.SendMsg(m)
}

func newWrappedStream(s grpc.ClientStream) grpc.ClientStream {
	return &wrappedSteam{s}
}

func clientStreamInterceptor(
	ctx context.Context,
	desc *grpc.StreamDesc,
	cc *grpc.ClientConn,
	method string,
	streamer grpc.Streamer,
	opts ...grpc.CallOption) (grpc.ClientStream, error) {
	s, err := streamer(ctx, desc, cc, method, opts...)
	if err != nil {
		return nil, err
	}
	return newWrappedStream(s), nil
}

func main() {
	conn, err := grpc.Dial(address,
		grpc.WithInsecure(),
		grpc.WithUnaryInterceptor(orderUnaryClientInterceptor),
		grpc.WithStreamInterceptor(clientStreamInterceptor))
	if err != nil {
		log.Fatalf("did not connect: %v", err)
	}
	defer conn.Close()
	c := pb.NewProductInfoClient(conn)
	name := "Apple iPhone 11"
	description := `Meet Apple iPhone 11. All-new dual-camera system with Ultra Wide and Night mode`

	md := metadata.Pairs(
		"timestamp", time.Now().Format(time.StampNano),
		"kn", "kv",
	)
	mdCtx := metadata.NewOutgoingContext(context.Background(), md)
	var header, trailer metadata.MD
	ctx, cancel := context.WithTimeout(mdCtx, time.Second*2)
	defer cancel()

	r, err := c.AddProduct(
		ctx,
		&pb.Product{Name: name, Description: description},
		grpc.Header(&header),
		grpc.Trailer(&trailer))
	if err != nil {
		log.Fatalf("Could not add product: %v", err)
	}
	fmt.Printf("header metadata: %v\n", header)
	fmt.Printf("trailer: %v\n", trailer)

	log.Printf("Product ID: %s added successfully", r.Value)
	product, err := c.GetProduct(ctx, r)
	if err != nil {
		log.Fatalf("Could not get product: %v", err)
	}
	log.Print("Product: ", product.String())

	oc := pb.NewOrderManagementClient(conn)
	order, err := oc.GetOrder(ctx, wrapperspb.String("test"))
	if err != nil {
		log.Printf("Could not get order: %v", err)
	}
	log.Printf("Order: %s", order.String())

	searchStream, _ := oc.SearchOrders(ctx, wrapperspb.String("test"))
	for {
		searchOrder, err := searchStream.Recv()
		if err == io.EOF {
			break
		}
		log.Print("Search Result : ", searchOrder)
	}
	/*
		{
			updateStream, err := oc.UpdateOrders(ctx)
			if err != nil {
				log.Fatalf("%v.UpdateOrders(_) = _, %v", oc, err)
			}
			if err := updateStream.Send(&pb.Order{Id: "1"}); err != nil {
				log.Fatalf("err : %v", err)
			}
			if err := updateStream.Send(&pb.Order{Id: "2"}); err != nil {
				log.Fatalf("err : %v", err)
			}
			if err := updateStream.Send(&pb.Order{Id: "3"}); err != nil {
				log.Fatalf("err : %v", err)
			}
			updateRes, err := updateStream.CloseAndRecv()
			if err != nil {
				log.Fatalf("err : %v", err)
			}
			log.Printf("Update Orders Res: %s", updateRes)
		}
		order, err = oc.GetOrder(ctx, wrapperspb.String("1"))
		if err != nil {
			log.Printf("Could not get order: %v", err)
		}
	*/
	streamProOrder, _ := oc.ProcessOrders(ctx)

	if err := streamProOrder.Send(wrapperspb.String("1")); err != nil {
		log.Fatalf("err : %v", err)
	}
	if err := streamProOrder.Send(wrapperspb.String("2")); err != nil {
		log.Fatalf("err : %v", err)
	}
	if err := streamProOrder.Send(wrapperspb.String("3")); err != nil {
		log.Fatalf("err : %v", err)
	}

	channel := make(chan struct{})

	go asncClientBidirectionRPC(streamProOrder, channel)
	if err := streamProOrder.Send(wrapperspb.String("4")); err != nil {
		log.Fatalf("err : %v", err)
	}
	fmt.Printf("streamProOrder header metadata: %v\n", header)
	fmt.Printf("streamProOrder trailer: %v\n", trailer)
	if err := streamProOrder.CloseSend(); err != nil {
		log.Fatalf("err : %v", err)
	}
	<-channel

	_, getOrderErr := oc.GetOrder(ctx, wrapperspb.String("test1111"))
	if getOrderErr != nil {
		errcode := status.Code(getOrderErr)
		if errcode == codes.NotFound {
			log.Printf("NotFound Error: %s", errcode)
			errStatus := status.Convert(getOrderErr)
			for _, d := range errStatus.Details() {
				switch info := d.(type) {
				case *epb.BadRequest_FieldViolation:
					log.Printf("Request Filed Invalid: %s", info)
				default:
					log.Print("Unexpected error type", info)
				}
			}
		} else {
			log.Printf("Unhandled error : %s ", errcode)
		}
	}
}

func asncClientBidirectionRPC(stream pb.OrderManagement_ProcessOrdersClient, c chan struct{}) {
	for {
		combinedShipment, err := stream.Recv()
		if err == io.EOF {
			break
		}
		//连接之后处理，似乎会阻塞
		header, _ := stream.Header()
		trailer := stream.Trailer()
		log.Printf("streamProOrder header metadata: %v\n", header)
		log.Printf("streamProOrder trailer: %v\n", trailer)
		log.Print("Combined shipment : ", combinedShipment.OrderList)
	}
	c <- struct{}{}
}
